package com.lzy.gmall.realtime.app.dim;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.lzy.gmall.realtime.utils.DateFormatUtil;
import com.lzy.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Random;

//9.1新老访客状态标记修复
public class BaseLogApp_g {
    public static void main(String[] args) throws Exception {
//        TODO 1. 初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//        TODO 2.设置并行度
        env.setParallelism(1);

//        TODO 3.从 Kafka 读取数据
        String topic="topic_log";
        String groupId="groupId";

//      Apache Flink 从 Kafka 主题中读取数据
        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

//        TODO 4.数据清洗，转换结构
        final SingleOutputStreamOperator<String> processed = source.process(
                new ProcessFunction<String, String>() {

                    @Override
                    public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
                        try {
                            out.collect(value);
                        } catch (Exception e) {
                            System.out.println(value);
                        }
                    }
                }
        );
//      4.3 转换主流数据结构 jsonStr -> jsonObj
        SingleOutputStreamOperator<JSONObject> mappedStream = processed.map(JSON::parseObject);

//        TODO 5.新老用户状态修复
//        按照mid 对数据进行分组
        KeyedStream<JSONObject, String> keyedStream = mappedStream
                .keyBy(r -> r.getJSONObject("common").getString("mid"));//mid相同的给共享

//        新老访客状态标记修复
        SingleOutputStreamOperator<JSONObject> fixedStream = keyedStream.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    //创建状态   给相同mid共享   容器里面只存日期   value  list   map
                    ValueState<String> firstViewDtState;


                    //open  ： 首先走open方法
                    @Override
                    public void open(Configuration param) throws Exception {
                        super.open(param);

                        firstViewDtState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstViewDtState",
                                String.class));
                    }
                    //zs   is_new=1    当前时间=dt       状态值=firstViewDt
                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        String isNew = jsonObj.getJSONObject("common").getString("is_new");
                        // 1   0


                        //状态时间
                        String firstViewDt = firstViewDtState.value();

                        //获取当前数据的时间
                        Long ts = jsonObj.getLong("ts");
                        //把时间戳转换成年月日
                        String dt = DateFormatUtil.toDate(ts);
                        if ("1".equals(isNew)) {
                            if (firstViewDt == null) {
                                firstViewDtState.update(dt);
                            } else {

                                if (!firstViewDt.equals(dt)) {
                                    isNew = "0";
                                    jsonObj.getJSONObject("common").put("is_new", isNew);
                                }
                            }
                        } else {
                            if (firstViewDt == null) {
                                // 将首次访问日期置为昨日
                                String yesterday = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                                firstViewDtState.update(yesterday);
                            }
                        }

                        out.collect(jsonObj);
                    }
                }
        );

//      TODO 6. 分流 page  start     err
        // 定义启动、曝光、动作、错误侧输出流
        OutputTag<String> startTag = new OutputTag<String>("startTag") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {
        };
        OutputTag<String> errorTag = new OutputTag<String>("errorTag") {
        };

        //新增字段shop_id
        SingleOutputStreamOperator<JSONObject> flatMapStrToJSON = fixedStream.flatMap(new FlatMapFunction<JSONObject, JSONObject>() {
            @Override
            public void flatMap(JSONObject value, Collector<JSONObject> out) throws Exception {
                // 1. 提取 common 对象
                JSONObject common = value.getJSONObject("common");

                // 2. 添加随机 shop_id（1-10）
                common.put("shop_id", new Random().nextInt(10) + 1);

                // 3. 添加随机 device_type（0 或 1）
                int i = new Random().nextInt(2);
                common.put("device_type", (i == 0 ? 0 : 1));

                // 4. 将修改后的对象输出
                out.collect(value);
            }
        });


//        分流
        SingleOutputStreamOperator<String> separatedStream = flatMapStrToJSON.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObj, Context context, Collector<String> out) throws Exception {
                        // 收集错误数据
                        JSONObject error = jsonObj.getJSONObject("err");
                        if (error != null) {
                            context.output(errorTag, jsonObj.toJSONString());
                        }
                        // 剔除 "err" 字段
                        jsonObj.remove("err");


                        // 收集启动数据
                        JSONObject start = jsonObj.getJSONObject("start");
                        if (start != null) {
                            context.output(startTag, jsonObj.toJSONString());
                        } else {
                            // 获取 "page" 字段
                            JSONObject page = jsonObj.getJSONObject("page");
                            // 获取 "common" 字段
                            JSONObject common = jsonObj.getJSONObject("common");
                            // 获取 "ts"
                            Long ts = jsonObj.getLong("ts");




                            // 收集曝光数据
                            JSONArray displays = jsonObj.getJSONArray("displays");
                            if (displays != null) {
                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject display = displays.getJSONObject(i); //[{},{},{}]
                                    //创建空的json对象
                                    JSONObject displayObj = new JSONObject();
                                    displayObj.put("display", display);
                                    displayObj.put("common", common);
                                    displayObj.put("page", page);
                                    displayObj.put("ts", ts);

                                    context.output(displayTag, displayObj.toJSONString());
                                }
                            }
                            // 收集动作数据
                            JSONArray actions = jsonObj.getJSONArray("actions");
                            if (actions != null) {
                                for (int i = 0; i < actions.size(); i++) {
                                    JSONObject action = actions.getJSONObject(i);
                                    JSONObject actionObj = new JSONObject();
                                    actionObj.put("action", action);
                                    actionObj.put("common", common);
                                    actionObj.put("page", page);
                                    actionObj.put("ts", ts);
                                    context.output(actionTag, actionObj.toJSONString());
                                }
                            }

                            // 收集页面数据
                            jsonObj.remove("displays");
                            jsonObj.remove("actions");



                            out.collect(jsonObj.toJSONString());
                        }

                    }
                }
        );






//        打印主流和各侧输出流查看分流效果
        separatedStream.getSideOutput(errorTag).print("error"); //错误
        separatedStream.getSideOutput(startTag).print("start");//启动
        separatedStream.getSideOutput(displayTag).print("display");//曝光
        separatedStream.getSideOutput(actionTag).print("action");//动作

//       TODO 7. 将数据输出到 Kafka 的不同主题
//       提取出输出流
        DataStream<String> startDS = separatedStream.getSideOutput(startTag);
        DataStream<String> displayDS = separatedStream.getSideOutput(displayTag);
        DataStream<String> actionDS = separatedStream.getSideOutput(actionTag);
        DataStream<String> errorDS = separatedStream.getSideOutput(errorTag);

//        // 7.2 定义不同日志输出到 Kafka 的主题名称
        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";

        separatedStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displayDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        errorDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));

        env.execute();
    }
}
